package com.zhaosc.rx.example;

import java.util.concurrent.TimeUnit;

import org.junit.Test;

import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.Completable;
import io.reactivex.CompletableEmitter;
import io.reactivex.CompletableObserver;
import io.reactivex.CompletableOnSubscribe;
import io.reactivex.CompletableSource;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import io.reactivex.MaybeEmitter;
import io.reactivex.MaybeOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.Single;
import io.reactivex.SingleEmitter;
import io.reactivex.SingleOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiConsumer;
import io.reactivex.functions.Cancellable;
import io.reactivex.functions.Consumer;
import io.reactivex.subscribers.DisposableSubscriber;

/**
 * RX2.X中Observable不再支持背压(Back Pressure)， 而改由Flowable来支持非阻塞式的背压。Flowable是RxJava
 * 2.x新增的被观察者， Flowable可以看成Observable新的实现，它支持背压，同时实现Reactive Streams的
 * Publisher接口。Flowable所有的操作符强制支持背压，不过幸运的是Flowable中的操作符 大多与Observable类似
 *
 * 使用 Flowable 较好的场景如下 : 1.处理以某种方式产生超过 lOKB 的元素; 2.文件读取与分析 ;
 * 3.读取数据库记录，也是一个阻塞的和基于拉取模式; 4.网络1/0流; 5.创建一个响应式非阻塞接口 。
 *
 * Rx.Java 2.x 中还有 3 种类型的被观察者 : Single、 Completable 和 Maybe。
 * 
 * @author zhaosc
 *
 */

public class OtherObservalExample {


	@Test
	public void testFlowable() {
		 
		Flowable.just("a", "b","c").forEach(t->System.out.println(t));
		Flowable.just("Hello world").subscribe(System.out::println);
		Disposable d = Flowable.just("Hello world!")
			     .delay(1, TimeUnit.SECONDS)
			     .subscribeWith(new DisposableSubscriber<String>() {
			         @Override public void onStart() {
			             System.out.println("Start!");
			         }
			         @Override public void onNext(String t) {
			             System.out.println(t);
			         }
			         @Override public void onError(Throwable t) {
			             t.printStackTrace();
			         }
			         @Override public void onComplete() {
			             System.out.println("Done!");
			         }
			     });
			SleepUtils.sleep(1000L);
			 d.dispose();
	}
	
	@Test
	public void testFlowable02() {
		String[] arr = { "a", "b", "c" };
		Flowable.fromArray(arr).subscribe(new Consumer<String>() {
			@Override
			public void accept(String s) {
				System.out.println("Hello " + s + "!");
			}
		});
	}
	
	/**
	 * single 只能发射一个数据
	 */
	@Test
	public void testSingle() {

		Single.create(new SingleOnSubscribe<String>() {

			@Override
			public void subscribe(SingleEmitter<String> emitter) throws Exception {
				emitter.onSuccess("zhaosc");//只能发射一个数据
				emitter.onSuccess("zhaosc2");
				emitter.onSuccess("zhaosc1");
			}
		}).subscribe(new BiConsumer<String, Throwable>() {

			@Override
			public void accept(String t, Throwable e) throws Exception {
				System.out.println("accept:" + t);
			}
		});

	}

	@Test
	public void testCompletable() {
		Completable completable = Completable.fromAction(() -> System.out.println("Hello World!"));
		completable.subscribe(() -> System.out.println("Done"), error -> error.printStackTrace());
	}

	@Test
	public void testFromRunnable() {

		Completable completable = Completable.fromRunnable(() -> {
			System.out.println("runnable");
		});
		completable.subscribe(() -> System.out.println("Done"), error -> error.printStackTrace());

	}
	

	@Test
	public void testCompletableEmitter() {

	 Completable.create(new CompletableOnSubscribe() {
			
			@Override
			public void subscribe(CompletableEmitter emitter) throws Exception {
				emitter.setCancellable(new Cancellable() {
					
					@Override
					public void cancel() throws Exception {
						System.out.println("****cancel****");
						
					}
				});
				emitter.onComplete();
			}
		}).andThen(new CompletableSource() {
			
			@Override
			public void subscribe(CompletableObserver cs) {
				cs.onComplete();
				System.out.println("*******subscribe*****");
				
			}
		}).andThen(Observable.rangeLong(1, 10)).subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
			System.out.println(t);
				
			}
		});

	}
	
	@Test
	public void testMaybe() {
		Maybe.create(new MaybeOnSubscribe<String> () {

			@Override
			public void subscribe(MaybeEmitter<String> emitter) throws Exception {
				emitter.onSuccess("hello");
				emitter.setDisposable(new Disposable() {
					
					@Override
					public boolean isDisposed() {
						return false;
					}
					
					@Override
					public void dispose() {
						System.out.println("****dispose****");
					}
				});
				
			}}).subscribe(new Consumer<String>() { 
			@Override
			public void accept(String t) throws Exception {
			System.out.println("hh:"+t);
			}});
		
	}

}
